package cn.kennylee.codehub.mq.lip.integration.rocketmq;

import cn.hutool.core.annotation.AnnotationUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.kennylee.codehub.mq.lip.integration.base.AbstractMqListener;
import cn.kennylee.codehub.mq.lip.integration.rocketmq.listener.AbstractRocketMqListener;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.lang.NonNull;
import org.springframework.util.StringUtils;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * <p> 动态RocketMqListener注册器 </p>
 * <p>使用BeanPostProcessor扩展接口进行扩展Bean，在Spring 容器完成 Bean 的实例化后执行自定义逻辑，每个bean都经过并且占用资源相对批量查询注册消耗少。 </p>
 * <p>Created on 2025/4/17.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@Slf4j
public class RocketMqListenerRegistrar implements BeanPostProcessor, ApplicationContextAware {

    public static final String CONSUMER_TLS_ENABLE_PLACEHOLDER = "${rocketmq.consumer.tls-enable:false}";

    private ApplicationContext applicationContext;

    private Environment env;

    /**
     * 消费者组集合
     */
    private final Set<String> consumerGroups = ConcurrentHashMap.newKeySet();

    /**
     * BeanPostProcessor的后置处理方法，在Bean实例化后执行
     *
     * @param bean     the new bean instance
     * @param beanName the name of the bean
     * @return the bean instance to use, either the original or a wrapped one; never {@code null}
     */
    @Override
    public Object postProcessAfterInitialization(@NonNull Object bean, @NonNull String beanName) {
        // 只支持单例Bean，否则会重复注册
        if (!applicationContext.containsBean(beanName) ||
            !applicationContext.isSingleton(beanName)) {
            return bean;
        }

        if (bean instanceof AbstractRocketMqListener abstractRocketMqListener) {
            // 如果是AbstractRocketMqListener的子类，实现了RocketMQMessageListener注解
            if (AnnotationUtil.hasAnnotation(AbstractRocketMqListener.class, RocketMQMessageListener.class)) {
                // 不任何操作
                return abstractRocketMqListener;
            }
            // 否则，注册RocketMqListener
            registerListener(abstractRocketMqListener);
        }
        return bean;
    }

    /**
     * 注册RocketMqListener
     *
     * @param listener RocketMqListener
     * @see ListenerContainerConfiguration#registerContainer(String, Object, RocketMQMessageListener)
     */
    @SneakyThrows
    private void registerListener(AbstractRocketMqListener listener) {

        final DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();

        // 设置消息监听器属性
        RocketMQMessageListener anno = listener.buildRocketMqMessageListener();
        if (log.isDebugEnabled()) {
            log.debug("RocketMQMessageListener annotation: {}", anno);
        }
        container.setRocketMQMessageListener(anno);
        // 设置RocketMQListener
        container.setRocketMQListener(listener);

        container.setName(listener.getInstanceName() + "-Container");

        container.setNameServer(env.resolveRequiredPlaceholders(anno.nameServer()));

        String accessChannel = env.resolveRequiredPlaceholders(anno.accessChannel());
        if (StringUtils.hasLength(accessChannel)) {
            container.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }

        // 设置消费组
        String consumerGroup = listener.isEnableGroupSuffix()
            // 如果启用消费组后缀，则添加环境后缀
            ? addRouterKeySuffix(listener.getConsumerGroup())
            : listener.getConsumerGroup();
        container.setConsumerGroup(consumerGroup);

        container.setTopic(env.resolvePlaceholders(listener.getTopic()));

        // 当前实例名字，方便在rocketmq控制台监控消费者。注意不要重复，否则影响调试，建议设置为类名
        container.setInstanceName(listener.getInstanceName());
        container.setTlsEnable(env.resolveRequiredPlaceholders(CONSUMER_TLS_ENABLE_PLACEHOLDER));
        container.setMessageConverter(applicationContext.getBean(RocketMQMessageConverter.class).getMessageConverter());

        String namespace = env.resolvePlaceholders(anno.namespace());
        container.setNamespace(RocketMQUtil.getNamespace(namespace,
            applicationContext.getBean(RocketMQProperties.class).getConsumer().getNamespace()));

        if (StrUtil.isBlank(listener.getSelectorExpression())) {
            container.setSelectorExpression(generateSelectorExpression(listener.getTags(), listener));
        } else {
            container.setSelectorExpression(addRouterKeySuffix(listener.getSelectorExpression()));
        }

        // 判断consumerGroups是否存在
        if (consumerGroups.contains(container.getConsumerGroup())) {
            log.error("消费者组已存在，消费者组：{}", container.getConsumerGroup());
            throw new IllegalArgumentException("消费者组已存在，消费者组：" + container.getConsumerGroup());
        } else {
            consumerGroups.add(container.getConsumerGroup());
        }

        if (log.isInfoEnabled()) {
            log.info("注册RocketMqListener：{}", container);
        }

        String containerBeanName = StrUtil.format("{}_{}", DefaultRocketMQListenerContainer.class.getName(),
            listener.getInstanceName());
        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
        genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
            () -> container);
    }

    /**
     * 生成消息选择表达式
     *
     * @param tags 消息标签集合
     * @return 消息选择表达式
     */
    private String generateSelectorExpression(Set<String> tags, AbstractMqListener<?> abstractMqListener) {
        if (CollUtil.isEmpty(tags)) {
            return "*";
        }
        if (abstractMqListener.isEnableTagSuffix()) {
            return tags.stream().map(this::addRouterKeySuffix).collect(Collectors.joining(" | "));
        } else {
            return CollUtil.join(tags, " | ");
        }
    }

    /**
     * 添加环境后缀
     *
     * @param s 目标字符串
     * @return 添加后缀的字符串
     */
    private String addRouterKeySuffix(String s) {
        return addRouterKeySuffix(s, applicationContext);
    }

    /**
     * 添加环境后缀
     *
     * @param s 目标字符串
     * @return 添加后缀的字符串
     */
    public static String addRouterKeySuffix(String s, ApplicationContext applicationContext) {
        // 获取当前首个profile作为后缀
        return StrUtil.trim(s) + "-" + applicationContext.getEnvironment().getActiveProfiles()[0];
    }

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        this.env = applicationContext.getEnvironment();
    }
}
